/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package com.mycompany.mqdemo3;

import cn.ac.iie.di.datadock.rdata.exchange.client.core.session.receive.REAbstractReceiveMessageHandler;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.ConsumePosition;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.connection.REConnection;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.REReceiveSession;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.REReceiveSessionBuilder;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 *发送消息，接收消息后，更改字段名转发给新的消息队列
 * 
 * @author dell
 */
public class Receive2 {
    //public static final String TOPIC="spftest22";
    public static final String TOPIC="6a3421eb345decbb";
    //public static final String  NAME_SERVER="192.168.203.128:9879";
   public static final String  NAME_SERVER="101.200.90.2:9876";
    public static final String GROUP="group_test_1";
    public static REReceiveSession session=null;
    public static void main(String[] args){
        
        REConnection conn=new REConnection(NAME_SERVER);
        REReceiveSessionBuilder builder=null;
        try {
            builder = (REReceiveSessionBuilder)conn.getReceiveSessionBuilder(TOPIC);
        } catch (Exception ex) {
           ex.printStackTrace();
        }       
        builder.setHandler(new REAbstractReceiveMessageHandler<REMessage>(){
            @Override
            //接收数据          
            public boolean handle(REMessage message) {
            try{        
             if(message==null){
                 //如果数据为空
                return false;
            }
             //读取数据
             for(Object[]obj:message.data){
                 HashMap map=new HashMap();
                 for(int i=0;i<message.colNames.length;i++){
                     if(obj[i] instanceof Integer){
                         map.put(message.colNames[i], obj[i]);
                     }else if(obj[i] instanceof String){
                         map.put(message.colNames[i],(String)obj[i].toString());
                     }else if(obj[i] instanceof Boolean){
                         map.put(message.colNames[i],obj[i]);
                     }else{
                         map.put(message.colNames[i],Arrays.toString((Map[])obj[i]));
                     }
                 }
                
                 System.out.println(map);
             }    
             }catch(Exception e){
                 e.printStackTrace();
             }
                return true;   
            }
        });
//        builder.setFailureHandler(new REAbstractReceiveMessageHandler<byte[]>() {
//            @Override
//            //处理错误数据
//            public boolean handle(byte[] message) {
//                System.out.println(message.length+"(||)"+message.toString());
//                return true;
//            }
//        });
        //设置读取数据的参数
        builder.setGroupName(GROUP);
        builder.setConsumPosition(ConsumePosition.CONSUME_FROM_LAST_OFFSET);
        builder.setConsumeThreadNum(10);
        try{       
        session=(REReceiveSession)builder.build();
        session.start();
        }catch(Exception e){
            e.printStackTrace();
        }
    }
}
